package heap;

import java.io.IOException;

import Transaction.Transaction;

import storage.data.Page;
import storage.data.PageID;

import Common.GlobalConst;
import Common.GoggleDB;
import exceptions.DBInvalidSlotNumberException;
import exceptions.HFBufMgrException;


public class Scan implements GlobalConst {
	/**
	 * The heapfile we are using
	 */
	private Heapfile hf;

	/** current dir page */
	private PageID dirpageId = new PageID();

	private HFPage dirpage = new HFPage();

	/**
	 * record ID of the DataPageInfo struct which describes the data page where
	 * our current record lives
	 */
	private RID datapageRID = new RID();

	// actual pageID
	private PageID datapageID = new PageID();

	private HFPage datapage = new HFPage();

	// current record from the current data page
	private RID userRID = new RID();

	/**
	 * status of next user status
	 */
	private boolean nextUserStatus;
	
	public Scan(Heapfile hf,Transaction tx) throws /*InvalidSlotNumberException,IO*/Exception {
		init(hf,tx);
	}

	private void pinPage(PageID pageno, Page page, boolean emptyPage)
			throws HFBufMgrException {
		try {
			GoggleDB.bufferMgr.pinPage(pageno, page, emptyPage);
		} catch (Exception e) {
			throw new HFBufMgrException(e);
		}

	}

	private void unpinPage(PageID pageno, boolean dirty)
			throws Exception {
		
		GoggleDB.bufferMgr.unpinPage(pageno, dirty);
		

	}
	
	/**
	 * Retrieve the next record in sequential scan
	 * @param rid
	 * @return
	 * @throws InvalidSlotNumberException
	 * @throws IOException
	 */
	public Tuple getNext(RID rid,Transaction tx,String table) throws DBInvalidSlotNumberException,IOException,HFBufMgrException,Exception {
		Tuple rectuple = null;
		
		if(nextUserStatus != true) {
			nextDataPage(tx,table);
		}
		
		if(datapage == null) {
			return null;
		}
		
		rid.pageNo.pageId = userRID.pageNo.pageId;
		rid.slotNo = userRID.slotNo;
		
		rectuple = datapage.getRecord(rid,tx,table);
		userRID = datapage.nextRecord(rid,tx,table);
		
		if(userRID==null) {
			nextUserStatus = false;
		} else {
			nextUserStatus = true;
		}
		
		return rectuple;
	}
	
	public void closeScan () {
		reset();
	}
	
	/**
	 * position the scan cursor to the record with the given rid
	 */
	
	public boolean position(RID rid,Transaction tx,String table) throws DBInvalidSlotNumberException,IOException,HFBufMgrException,Exception {
		RID nextRID = new RID ();
		boolean bst;
		bst = true;
		nextRID.pageNo.pageId = userRID.pageNo.pageId;
		nextRID.slotNo = userRID.slotNo;
		
		if(nextRID.equals(rid)==true) {
			return true;
		}
		
		PageID pgid = new PageID ();
		pgid.pageId = rid.pageNo.pageId;
		
		if(!datapageID.equals(pgid)) {
			reset ();
			bst = firstDataPage(tx,table);
			if(bst != true) {
				return bst;
			}
			
			while(!datapageID.equals(rid)) {
				bst = nextDataPage(tx,table);
				if(bst!=true) {
					return bst;
				}
			}
		}
		//Now in the correct page
		userRID = datapage.firstRecord(tx,table);
		if(userRID == null) {
			bst = false;
			return bst;
		}
		
		bst = true;
		nextRID.pageNo.pageId = userRID.pageNo.pageId;
		nextRID.slotNo = userRID.slotNo;
		
		while((bst == true) && (nextRID!=rid)) {
			bst = mvNext(nextRID,tx,table);
		}
		
		return bst;
	}
	
	//move to next record and return the current new rid
	
	private boolean mvNext(RID rid,Transaction tx,String table) throws DBInvalidSlotNumberException,IOException,HFBufMgrException ,Exception{
		RID nextRID;
		boolean status;
		
		if(datapage==null) {
			return false;
		}
		
		nextRID = datapage.nextRecord(rid,tx,table);
		
		if(nextRID != null) {
			userRID.pageNo.pageId = nextRID.pageNo.pageId;
			userRID.slotNo = nextRID.slotNo;
			return true;
		} else {
			status = nextDataPage(tx,table);
			
			if(status==true) {
				rid.pageNo.pageId = userRID.pageNo.pageId;
				rid.slotNo = userRID.slotNo;
			}
		}
		return true;
	}
	
	private void reset () {
		if(datapage!=null) {
			try {
				unpinPage(datapageID,false);
			}catch (Exception e) {
				e.printStackTrace();
			}
			
		}
		datapageID.pageId = 0;
		datapage = null;
		if(dirpage!= null) {
			try {
				unpinPage(dirpageId,false);
			}catch (Exception e) {
				e.printStackTrace();
			}
		}
		dirpage = null;
		nextUserStatus = true;
	}

	private boolean firstDataPage(Transaction tx,String table) throws IOException, HFBufMgrException,
			DBInvalidSlotNumberException ,Exception{
		DataPageInfo dpinfo;
		Tuple rectuple = null;
		boolean bst;

		dirpageId.pageId = hf.firstDirPageID.pageId;
		nextUserStatus = true;
		
		dirpage = new HFPage();
		pinPage(dirpageId, (Page) dirpage, false);
		datapageRID = dirpage.firstRecord(tx,table);
		if (datapageRID != null) {
			rectuple = dirpage.getRecord(datapageRID,tx,table);
			dpinfo = new DataPageInfo(rectuple,tx,table);
			datapageID.pageId = dpinfo.pageId.pageId;
		} else {
			PageID nextDirPageID = new PageID();
			nextDirPageID = dirpage.getNextPage(tx,table);

			if (nextDirPageID.pageId != INVALID) {
				
				unpinPage(dirpageId, false);
				
				dirpage = null;
				dirpage = new HFPage();
				pinPage(nextDirPageID, (Page) dirpage, false);
				datapageRID = dirpage.firstRecord(tx,table);
				if (datapageRID != null) {
					rectuple = dirpage.getRecord(datapageRID,tx,table);
					if (rectuple.getLength() != DataPageInfo.size) {
						return false;
					}
					dpinfo = new DataPageInfo(rectuple,tx,table);
					datapageID.pageId = dpinfo.pageId.pageId;
				} else {
					datapageID.pageId = INVALID;
				}
			} else {
				datapageID.pageId = INVALID;
			}
		}
		//System.out.println("datapageprev "+datapageID.pageId);
		datapage = null;
		nextDataPage(tx,table);
		return true;
	}

	private boolean nextDataPage(Transaction tx,String table) throws IOException, HFBufMgrException,
			DBInvalidSlotNumberException ,Exception{
		DataPageInfo dpinfo;
		
		boolean nextDataPageStatus;
		PageID nextDirPageID = new PageID ();
		Tuple rectuple = null;
		if((dirpage==null) && (datapageID.pageId==INVALID)) {
			return false;
		}

		//System.out.println("datapageID "+datapage);
		if(datapage==null) {
			if(datapageID.pageId==INVALID) {
				//heapfile is empty
				//System.out.println("datapageID "+dirpageId.pageId);
				//System.out.println("pindirpageid "+dirpageId.pageId);
				unpinPage(dirpageId,false);
				//System.out.println("dirpageid "+dirpageId.pageId);
				dirpage = null;
			} else {
				datapage = new HFPage();
				pinPage(datapageID,(Page)datapage,false);
				userRID = datapage.firstRecord(tx,table);
				return true;
			}
		}
		//System.out.println("datapageID "+datapageID.pageId);
		if(datapageID.pageId!=INVALID) {
			unpinPage(datapageID,false);
		}
		datapage = null;
		if(dirpage == null) {
			return false;
		}
		datapageRID = dirpage.nextRecord(datapageRID,tx,table);
		
		if(datapageRID==null) {
			nextDataPageStatus = false;
			nextDirPageID = dirpage.getNextPage(tx,table);
			unpinPage(dirpageId,false);
			dirpage = null;
			datapageID.pageId = INVALID;
			if(nextDirPageID.pageId==INVALID) {
				return false;
			} else {
				dirpageId = nextDirPageID;
				dirpage = new HFPage();
				pinPage(dirpageId,(Page)dirpage,false);
				if(dirpage==null) {
					return false;
				}
				datapageRID = dirpage.firstRecord(tx,table);
				nextDataPageStatus=true;
				
			}
		}
		rectuple = dirpage.getRecord(datapageRID,tx,table);
		if(rectuple.getLength()!= DataPageInfo.size) {
			return false;
		}
		
		dpinfo = new DataPageInfo(rectuple,tx,table);
		datapageID.pageId = dpinfo.pageId.pageId;
		datapage = new HFPage();
		pinPage(dpinfo.pageId,(Page)datapage,false);
		userRID = datapage.firstRecord(tx,table);
		
		if(userRID == null) {
			nextUserStatus = false;
			return false;
		}
		
		return true;
		
	}

	private void init(Heapfile hfile,Transaction tx) throws Exception {
		hf = hfile;
		firstDataPage(tx,hfile.fileName);
	}
}
